package com.baixiaowen.disruptor.Disruptor快速使用;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Disruptor 核心组建
 *
 * @author oker
 */
public class Main {

    public static void main(String[] args) {

        // 参数准备工作
        OrderEventFactory orderEventFactory = new OrderEventFactory();
        int ringBufferSize = 1024 * 1024;
        // Runtime.getRuntime().availableProcessors()
        ExecutorService executor = Executors.newFixedThreadPool(2);

        /**
         * 1.eventFactory   - 消息工厂对象
         * 2.ringBufferSize - 容器的长度
         * 3.executor       - 线程池(建议使用自定义线程池)
         * 4.producerType   - 生产者类型 单生产者:多生产者
         * 5.waitStrategy   - 等待策略
         */
        // 1. 实例化disruptor对象
        Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(
                orderEventFactory,
                ringBufferSize,
                executor,
                ProducerType.SINGLE, // 单生产者模式
                new BlockingWaitStrategy() // 阻塞的等待策略
                );

        // 2. 添加消费者监听 (disruptor 与 消费者的一个关联关系)
        disruptor.handleEventsWith(new OrderEventHandler(), new OrderEventHandler(), new OrderEventHandler());

        // 3. 启动Disruptor
        disruptor.start();

        // 4. 获取实际存储数据的容器: RingBuffer
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        OrderEventProducer producer = new OrderEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);

        for (long i = 0; i < 100; i ++) {
            bb.putLong(0, i);
            producer.sendData(bb);
        }

        disruptor.shutdown();
        executor.shutdown();

    }

}
